/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.pig.pigpen.util;

import java.io.File;
import java.net.URL;
import java.security.CodeSource;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import org.apache.pig.impl.util.PropertiesUtil;
import org.apache.pig.pigpen.ConsoleLogger;
import org.apache.pig.pigpen.PigPenActivator;
import org.apache.pig.pigpen.LoggerUtil;
import org.apache.pig.pigpen.PluginConfiguration;
import org.apache.pig.pigpen.preferences.PreferenceConstants;
import org.eclipse.core.resources.IFile;
import org.eclipse.core.runtime.Platform;
import org.eclipse.jface.preference.IPreferenceStore;
import org.eclipse.swt.widgets.Text;
import org.osgi.framework.Bundle;

public class PigLauncher implements Runnable {

    static final boolean forceLocalExec = true; // helps with debugging

    IFile file;
    Text output;

    Process proc = null;

    ConsoleLogger logger = ConsoleLogger.getDefault();

    public PigLauncher(IFile file, Text output) {
        this.file = file;
        this.output = output;

        logger.clear();
        logger.setFocus();
    }

    public void run() {

        try {
            // run Pig in separate process (don't want to run it in the UI
            // thread!)

            logger.error("Launching the job!");
            List<String> args = new LinkedList<String>();

            File jvm = new File(
                    new File(System.getProperty("java.home"), "bin"), "java");
            args.add(jvm.toString());

            String pigLocation = getPigLocation();
            String sep = System.getProperty("path.separator");
            StringBuffer classPath = new StringBuffer();
            classPath.append(System.getProperty("java.class.path"));
            if (pigLocation != null) {
                classPath.append(sep);
                classPath.append(pigLocation);
            }
            args.add("-classpath");
            IPreferenceStore store = PigPenActivator.getDefault()
                    .getPreferenceStore();
            String stringList = store
                    .getString(PreferenceConstants.P_PROPERTIES);
            Map<String, String> PrefProperties = PluginConfiguration
                    .parseString(stringList);
            String confPath = PrefProperties
                    .get(PreferenceConstants.P_CONFPATH);
            if (confPath == null || confPath.length() == 0) {

                logger
                        .debug("Please set the variable "
                                + PreferenceConstants.P_CONFPATH
                                + " to point to the directory containing the configuration files in Pig preferences");
                return;
            }
            logger.info("Using the configuration from " + confPath);

            args.add(classPath.toString() + sep + confPath);

            // String sshGateway =
            // PrefProperties.get(PreferenceConstants.P_SSHGATEWAY);
            Properties properties = PropertiesUtil.loadPropertiesFromFile();
            String sshGateway = properties.getProperty("ssh.gateway");

            if (sshGateway != null && !"".equals(sshGateway) /*
                                                              * sshGateway.length
                                                              * () > 0
                                                              */) {
                logger.info("Using ssh.gateway = " + sshGateway);
                args.add("-Dssh.gateway=" + sshGateway);
            }

            // args.add(classPath.toString() + ":" +
            // "/home/shubhamc/workspace/eclipsePlugin/conf");
            args.add("org.apache.pig.Main");

            args.add("-exectype");
            args.add("mapreduce");

            String filename = file.getRawLocation().toPortableString();// +
            // "tmp."
            // +
            // file.
            // getRawLocation
            // ().
            // getFileExtension
            // ();
            // String filename = file.getPath();

            args.add("-f");
            args.add(filename);

            ProcessBuilder pb = new ProcessBuilder(args);
            String parentDir = file.getRawLocation().removeLastSegments(1)
                    .toPortableString();
            // String parentDir = file.getParent();
            pb.directory(new File(parentDir)); // set process working directory
            // to directory that the .pig
            // file is in
            proc = pb.start();

            // redirect stdout and stderr to us
            StreamGobbler errorGobbler = new StreamGobbler(proc
                    .getErrorStream(), this.output, true);
            StreamGobbler outputGobbler = new StreamGobbler(proc
                    .getInputStream(), this.output, false);

            // kick off the stream gobblers in separate threads
            /*
             * output.getDisplay().asyncExec(errorGobbler);
             * output.getDisplay().asyncExec(outputGobbler);
             */
            Thread errorT = new Thread(errorGobbler);
            Thread outputT = new Thread(outputGobbler);
            errorT.start();
            outputT.start();

        } catch (Throwable t) {
            LoggerUtil.logError("Error launching Pig", t);
        }
    }

    public void kill() {
        if (proc != null) {
            proc.destroy(); // TODO: ask Pig to exit gracefully
            proc = null;

            /*
             * if(file.delete()) return; else
             * PigPenLog.logError("Error deleting temporary file" +
             * file.getPath());
             */
        }
    }

    // private String getPigLocation() throws Exception {
    //		
    // /*
    // String pigJar =
    // file.getWorkspace().getRoot().getLocation().append("pig.jar"
    // ).toPortableString();
    //		
    // File f = new File(pigJar);
    // if (f.exists()) return pigJar;
    // else throw new Exception(
    // "Unable to find pig.jar to execute; please put pig.jar at workspace root"
    // );
    // */
    //		
    // String sep = System.getProperty("path.separator");
    // StringBuffer sb = new StringBuffer();
    //		
    // Bundle plugin = Platform.getBundle("org.apache.pig.pigpen");
    //		
    // URL url = plugin.getEntry("/lib/pig.jar");
    // if (url != null) {
    // sb.append(resolvePath(url));
    // }
    //		
    // // also try the plugin installation directory
    // url =
    // Platform.getPlugin("org.apache.pig.pigpen").getDescriptor().getInstallURL
    // ();
    // if (url != null) {
    // if (sb.length() > 0) sb.append(sep);
    // sb.append(resolvePath(url));
    // }
    //		
    // return sb.toString();
    // }
    //
    // private String resolvePath(URL url) throws Exception {
    // URL resolvedURL = Platform.resolve(url);
    // File file = new File (resolvedURL.getFile());
    // String path = file.getAbsolutePath();
    // if (path.endsWith("!")) path = path.substring(0, path.length()-1);
    // return path;
    // }

    private String getPigLocation() {
        String ret;
        CodeSource source = org.apache.pig.Main.class.getProtectionDomain()
                .getCodeSource();

        if (source != null) {

            ret = source.getLocation().getPath();

        } else {
            logger.error("org.apache.pig.Main doesn't exist");
            ret = null;
        }
        return ret;
    }

    public synchronized void outputLine(String line) {
        output.insert(line);
        output.insert("\n");

    }

}